package com.dongdongshop.mq;

import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.pojo.TbUserScoreDetail;
import com.dongdongshop.pojo.TbUserScoreUse;
import com.dongdongshop.service.ScoreAccountService;
import com.dongdongshop.service.ScoreDetailService;
import com.dongdongshop.service.ScoreUseService;
import io.netty.channel.sctp.SctpChannelOption;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.mail.MailProperties;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
@Component
@RocketMQMessageListener(consumerGroup = "get_socre",topic = "create_order")
public class ScoreConsumer implements RocketMQListener<String> , RocketMQPushConsumerLifecycleListener {


    private Logger logger = LoggerFactory.getLogger(ScoreConsumer.class);
    @Resource
    private ScoreAccountService scoreAccountService;
    @Resource
    private ScoreDetailService scoreDetailService;


    /*onmessage相当于默认配置,使用preaprestart后将失效*/
    @Override
    public void onMessage(String stringObjectMap) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                MessageExt msg = list.get(0);
                try {
                    //解决幂等性问题
                    String msgbody = new String(msg.getBody(), "utf-8");
                    Map map = JSONObject.parseObject(msgbody, Map.class);
                    BigDecimal accepPass = (BigDecimal)map.get("total");
                    Long total = accepPass.longValue();
                    String userId = (String) map.get("userId");

                    /*添加积分业务*/
                    TbUserScoreAccount tbUserScoreAccount = scoreAccountService.getByUserId(userId);
                    Date date = new Date();
                    if (tbUserScoreAccount == null){
                        TbUserScoreAccount scoreAccount = new TbUserScoreAccount();
                        scoreAccount.setUserId(userId);
                        scoreAccount.setTotalScore(total);
                        scoreAccount.setExpendScore(0L);
                        scoreAccount.setCreatetime(date);
                        scoreAccount.setUpdatetime(date);
                        scoreAccountService.addinfo(scoreAccount);
                    }else{
                        tbUserScoreAccount.setTotalScore(tbUserScoreAccount.getTotalScore()+total);
                        tbUserScoreAccount.setUpdatetime(date);
                        scoreAccountService.updateinfo(tbUserScoreAccount);
                    }
                    logger.info("添加积分表成功");
                    /*添加积分流水表*/
                    TbUserScoreDetail tbUserScoreDetail = new TbUserScoreDetail();
                    tbUserScoreDetail.setUserId(userId);
                    tbUserScoreDetail.setScore(total);
                    tbUserScoreDetail.setAddtime(date);
                    Calendar c = Calendar.getInstance();
                    c.setTime(date);
                    c.add(Calendar.DATE,7);
                    Date closeTime = c.getTime();
                    tbUserScoreDetail.setClosetime(closeTime);
                    scoreDetailService.addinfo(tbUserScoreDetail);
                    logger.info("积分流水表添加完成");

                } catch (Exception e) {
                    e.printStackTrace();
                    int reconsumeTimes = msg.getReconsumeTimes();
                    // 该条消息可以存储到DB或者LOG日志中，或其他处理方式，然后重试
                    if (reconsumeTimes >= 2) {
                        System.out.println("已经重试三次,停止重试,将数据放入私信队列");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
